package com.qingcloud.df.executor.queue.subscriber;

import akka.actor.*;
import cn.hutool.extra.spring.SpringUtil;
import com.qingcloud.df.executor.eventbus.MessageEventBus;
import com.qingcloud.df.executor.eventbus.MessageEventSubActor;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.transport.akka.AkkaInitializer;
import com.qingcloud.df.sdk.component.transport.engine.RemoteExeEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;

/**
 * akka订阅处理类
 */
public class AkkaSubcriber implements ComponentSubscriber {
    Logger logger =  LoggerFactory.getLogger(AkkaSubcriber.class);
    /**
     * 事件总线
     */
    private final MessageEventBus messageEventBus;

    String actorNameTemplate  = "actor_%s_%s_%s";  //actor_dev_1109_1211  or  actor_prod_1109_1211
    static private HashMap<String, ActorRef> refs = new HashMap<>();

    static public ActorRef getActorRef(String actorName) {
        return refs.get(actorName);
    }

    public AkkaSubcriber(){
        messageEventBus = SpringUtil.getBean(MessageEventBus.class);
    }
    /**
     * 订阅主题
     * @param topics 主题列表
     */
    @Override
    public void subscribe(CompInstContext context ,List<String> topics) {
        if(topics!=null && topics.size()>0){
            ActorSystem actorSystem = RemoteExeEngine.getExecutorActorSystem();

            String env = context.getEnv();
            Integer jobId=context.getJobId();
            Integer instanceId =  context.getId();
            String actorName = String.format(actorNameTemplate,env,jobId,instanceId);
            ActorRef actorRef = null;
            if(refs== null || getActorRef(actorName)==null) {
                actorRef = actorSystem.actorOf(Props.create(MessageEventSubActor.class, context), actorName);
                refs.put(actorName,actorRef);
                logger.debug("add new actorRef-{} successful",actorName);
            }else {
                actorRef = getActorRef(actorName);
            }

            for(String topic : topics){
                try {
                    messageEventBus.subscribe(actorRef, topic);
                    logger.debug("subscirbe topic 【{}】 successful",topic);
                }catch (Exception e){
                    logger.error("-----------》subscirbe topic 【{}】 error《----------", topic);
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public void remove(Integer jobId,Integer instanceId, String env){
        String actorName = String.format(actorNameTemplate,env,jobId,instanceId);
        refs.entrySet().removeIf(entry -> (actorName==entry.getKey()));
    }


    /**
     * 取消订阅
     */
    @Override
    public void unsubscribe(CompInstContext context ,List<String> topics) {
        if(topics !=null && topics.size()>0){
            String env = context.getEnv();
            Integer jobId=context.getJobId();
            Integer instanceId =  context.getId();
            String actorName = String.format(actorNameTemplate,env,jobId,instanceId);
            ActorRef actorRef = getActorRef(actorName);
            if(actorRef!=null) {
                for (String topic : topics) {
                    messageEventBus.unsubscribe(actorRef, topic);
                }
            }
            remove(jobId,instanceId,env);
        }
    }
}
